//
// Created by forec on 17-5-8.
//

#include "../include/client.h"
#include <gflags/gflags.h>
#include "sys/time.h"
#include <stdint.h>
#include <unistd.h>
#include <chrono>


using namespace chrono;

using namespace std;

int MaxMessage  = 10000;

DEFINE_string(maddr, "192.168.50.10", "Master address, default to localhost.");
DEFINE_int32(mport, MASTER_PORT, ("Master port, default to " + std::to_string(MASTER_PORT) + ".").c_str());
DEFINE_int32(requests, 50000, "Total number of requests, default to 5000.");
DEFINE_int32(w, 100, "Percentage of updates(writes), default to 100.");
DEFINE_int32(r, 1, "Split the total number of requests into this many rounds, and do rounds "
    "sequentially, default to");
DEFINE_int32(eps, 0, "Send eps more messages per round than the client will wait for (to discount "
    "stragglers), default to 0.");
DEFINE_int32(conflicts, 0, "Percentage of conflicts, default to 0%.");
DEFINE_int32(segment, 10, "The segment count of delay distribution.");
DEFINE_int32(mins, MIN_VALUE_SIZE, "Minimum value size of a proposal.");
DEFINE_int32(maxs, MAX_VALUE_SIZE, "Maximum value size of a proposal.");
DEFINE_bool(check, false, "Check that every expected reply was received exactly once.");
DEFINE_bool(delay, false, "Get the distribution of single delay.");
DEFINE_bool(debug, false, "Output debug msg.");
DEFINE_validator(mport, &ValidatePort);
DEFINE_validator(conflicts, &ValidatePercent);
DEFINE_validator(w, &ValidatePercent);
DEFINE_validator(mins, &ValidateSize);
DEFINE_validator(maxs, &ValidateSize);

unsigned long N;
std::vector<int> successful;
std::vector<int> rarray;
std::vector<bool> rsp;
std::vector<clock_t> delay;


// int client_send_message(rdma_fd *handler, serializeBuff *message, TYPE msg_type){
//     int ret = 0;
// //    query_qp(handler);
//     int send_buf_size = sizeof(uint8_t) + sizeof(uint32_t) + 1 + message->len + 1;
//     char *send_buf = (char *)malloc(send_buf_size);
//     gen_send_buf(send_buf, message, msg_type);
//     ret = rdma_write(handler, send_buf, send_buf_size);
//     free(send_buf);
//     return ret;
// }


int client_send_message(rdma_fd *handler, serializeBuff *message, TYPE msg_type){
    int ret = 0;
//    query_qp(handler);
    uint8_t type = (uint8_t)msg_type;
    char *buf_one = (char *)malloc(sizeof(uint8_t) + sizeof(uint32_t) + 1);
    memcpy(buf_one, &type, sizeof(uint8_t));
    memcpy(buf_one + sizeof(uint8_t), &message->off, sizeof(uint32_t));
    *(buf_one + 5) = '1';
    ret = rdma_write(handler, buf_one, 6);
    if(ret == -1){
        return ret;
    }
//    printf("message->len: %d\n", message->len);
    free(buf_one);
    char *buf_two = (char *)malloc(message->off + 1);
    memcpy(buf_two, message->buff, message->off);
    *(buf_two + message->off) = '1';
    ret = rdma_write(handler, buf_two, message->off + 1);
    free(buf_two);
    return ret;
}
void latencyTest(rdma_fd *handler){
    struct timeval tv_begin, tv_end;
    int64_t total = 0;
    int messageCount = MaxMessage;
    uint8_t msg_type = 0;
    auto endTime = std::chrono::system_clock::now();
    //init propose
        
    //send 10000 messages and calculate the latency
    for(int i = 0;i < MaxMessage;i++){
        char * buf = new char[10];
        memset(buf, 0, (size_t)10);
        serializeBuff *replymessage;
        //init tk_command
        tk_command * command = new tk_command();
        command->opcode = PUT;
        command->key = "1234";
        command->valSize = 10;
        command->val = buf;

        Propose *propose = new Propose();
        propose->Command = *command;
        //send message
        auto startTime = std::chrono::system_clock::now();
        serializeBuff * message = create_serialize_buff();
        serializePropose(propose,message);
        client_send_message(handler,message,PROPOSE);

        int32_t len = 0;
        //the time it send the message
        //receive message
        //gettimeofday(&tv_begin, NULL);
        
        while(1){
            if(handler->have_read + sizeof(uint8_t) + sizeof(uint32_t) + 1 > handler->receive_buf_size){
                handler->have_read = 0;
            }

            while(1){
                volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + 5);
                if(*end == '1'){
                    break;
                }
            }
        
        
        //the time it received a response
        //memcpy(&msg_type, handler->receive_buf + handler->have_read, sizeof(uint8_t));
        memcpy(&len, handler->receive_buf + handler-> have_read + sizeof(uint8_t), sizeof(uint32_t));
        //memset(handler->receive_buf + handler->have_read, 0, sizeof(uint8_t) + sizeof(uint32_t) + 1);
        handler->have_read += sizeof(uint8_t) + sizeof(uint32_t) + 1;
        if(handler->have_read + len + 1 > handler->receive_buf_size){
            handler->have_read = 0;
        }
        while(1){
            volatile char *end = (volatile char *)(handler->receive_buf + handler->have_read + len);
            if(*end == '1'){
                break;  
            }
        }
        endTime = std::chrono::system_clock::now();
        //printf("2.have receive msg from id: %d, msg len: %d\n", remote_id, message->len);
        char * buff = (char *)malloc(len);
        memcpy(buff, handler->receive_buf + handler->have_read, len);
        replymessage = create_serialize_buff(buff, len);
        memset(handler->receive_buf + handler->have_read, 0, len + 1);
        handler->have_read += len + 1;
//        printf("before dispatch, message: %p, message->len: %d\n", message, message->len);
//        r->msg_dispatcher(remote_id, message, (TYPE)msg_type);
        int ret = 0;
        ProposeReply *propose_reply = new ProposeReply();
        ProposeReplyTS *propose_reply_ts = new ProposeReplyTS();
        if((TYPE)msg_type == PROPOSE_REPLY){
            ret = deserializeProposeReply(replymessage, propose_reply);
            if(ret == -1){
                continue;
                printf("deserialize proposereply failed\n");
            }
            else{
                if(propose_reply->OK){

                }
                // fprintf(stdout, "i = %d, Receive reply from replica %d, withhandleCommitShort command = %d\n",
                //     i, leader, propose_reply->CommandId);
            }
        }
        else if((TYPE)msg_type == PROPOSE_REPLY_TS){
            ret = deserializeProposeReplyTS(replymessage, propose_reply_ts);
            if(ret == -1){
                continue;
                printf("deserialize proposereply ts failed\n");
            }
            else{
                if(propose_reply_ts->OK){

                }
            //     fprintf(stdout, "i = %d, Receive reply from replica %d, withhandleCommitShort command = %d\n",
            //         i, leader, propose_reply_ts->CommandId);
            }
        }
        delete propose_reply;
        delete propose_reply_ts;
        break;
        }
        
        delete message;
        //delete replymessage;
        //
        
        //count time
        auto duration = duration_cast<microseconds>(endTime - startTime);
        //int64_t thistime = tv_end.tv_usec - tv_begin.tv_usec + (int64_t)(tv_end.tv_sec - tv_begin.tv_sec) * 1000000;
        //cout<<"thistime= "<<thistime<<endl;
        //if((int64_t)duration.count() < 300){
            total += (int64_t)duration.count();
            //messageCount--;
        //}
        cout<<"send "<<i<<"th message, the latency is "<<(int64_t)duration.count()<<"us"<<endl;
    }
    
    //total = tv_end.tv_usec - tv_begin.tv_usec + (int64_t)(tv_end.tv_sec - tv_begin.tv_sec) * 1000000;
    cout<<"total time="<<total<<endl;
    //cout<<"messageCount = "<<messageCount<<endl;
    double latency = (double)total / MaxMessage;
    printf("send 10000 message complete! average latency = %lfus\n",latency);
}
int main(int argc, char * argv[]) {
    gflags::ParseCommandLineFlags(&argc, &argv, true);

    int master = dialTo(FLAGS_maddr, FLAGS_mport);
    if (master < 0) {
        fprintf(stderr, "Error connecting to master!\n");
        exit(1);
    }

    uint8_t msgType;
    GetReplicaListArgs rl;
    GetReplicaListReply rlp;
    rl.Marshal(master);
    if (readUntil(master, (char *) &msgType, 1) != 0 || (TYPE)msgType != GET_REPLICA_LIST_REPLY ||
        !rlp.Unmarshal(master)) {
        fprintf(stderr, "Error when getting replica lists from master!\n");
        exit(1);
    }
    if (!rlp.Ready) {
        fprintf(stderr, "Master's reply shows that replicas are not ready!\n");
        exit(0);
    }

    //N = (unsigned long)rlp.ReplicaAddrList.size();
    N = 1;
    successful.resize(N, 0);
    std::vector<int> servers(N, -1);
    std::vector<rdma_fd *> servers_rdma(N, NULL);
    std::vector<std::string> karray((unsigned long)FLAGS_requests / FLAGS_r + FLAGS_eps, "");
    std::vector<int> rarray((unsigned long)FLAGS_requests / FLAGS_r + FLAGS_eps, -1);
    std::vector<bool> put((unsigned long)FLAGS_requests / FLAGS_r + FLAGS_eps, true);
    std::vector<int> perReplicaCount(N, 0);
    std::unordered_map<std::string, int> test;
    context_info *rdma_ib_info = (context_info *)malloc(sizeof(context_info));
    open_device_and_alloc_pd(rdma_ib_info);
    srand((unsigned int)time(nullptr));

    for (int i = 0; i < rarray.size(); i++) {
        int r = rand() % N;
        rarray[i] = r;
        if (i < FLAGS_requests / FLAGS_r) {
            perReplicaCount[r] ++;
        }
        if (FLAGS_conflicts > 0) {
            r = rand() % 100;
            if (r < FLAGS_conflicts) {
                karray[i] = CONFLICT_KEY;
            } else {
                karray[i] = TEST_KEY + std::to_string(i);
            }
            r = rand() % 100;
            put[i] = false;
            if (r < FLAGS_w)
                put[i] = true;
        } else {
            karray[i] = TEST_KEY + std::to_string(i);
            test[karray[i]] ++;
        }
    }

    if (FLAGS_conflicts > 0) {
        fprintf(stdout, "Conflict = %d%% distribution\n", FLAGS_conflicts);
    } else {
        fprintf(stdout, "Non-same distribution\n");
    }

    if (FLAGS_delay) {
        delay.resize((unsigned long)FLAGS_requests, 0);
    }

    for (int i = 0; i < N; i++) {
        servers[i] = dialTo(rlp.ReplicaAddrList[i], (uint16_t )rlp.ReplicaPortList[i]);
        cout<<rlp.ReplicaAddrList[i]<<endl;
        if (servers[i] < 0) {
            fprintf(stderr, "Error connecting to replica %d\n", i);
            continue;
        }

        int on = sizeof(unsigned int);
        setsockopt(servers[i], SOL_SOCKET, SO_KEEPALIVE, (void *)&on, on);  // make socket keeps alive
        setsockopt(servers[i], IPPROTO_TCP, TCP_NODELAY, (void *)&on, on);  // forbids NAGLE algorithm
        rdma_fd *handler = (rdma_fd *)malloc(sizeof(rdma_fd));
        handler->fd = servers[i];               
        get_context_info(handler, rdma_ib_info);
        build_rdma_connection(handler);
        printf("connection complete!\n");
        servers_rdma[i] = handler;
    }

    fprintf(stdout, "All replica connected, starting proposal sending...\n");
    //printf("tk_command() size : %d\n", sizeof(tk_command));
    int32_t id = 0;
    MsgQueue * mq = new MsgQueue(N, 1);
    tk_command tkc;
    Propose args(id, 0, tkc, -1);

    char * buf = new char[FLAGS_maxs + 1];
    memset(buf, 0, (size_t)FLAGS_maxs + 1);

    std::string _bufBegin = randStr(FLAGS_mins, FLAGS_mins);
    strcpy(buf, _bufBegin.c_str());

    int32_t bufIndex = FLAGS_mins;
    int32_t msize = FLAGS_maxs;
    while (msize > 0) {
        if (msize > 1024) {
            msize -= 1024;
            std::string _v = randStr(1024, 1024);
            strcpy(buf + bufIndex, _v.c_str());
            bufIndex += 1024;
        } else {
            std::string _v = randStr(msize, msize);
            strcpy(buf + bufIndex, _v.c_str());
            break;
        }
    }
        
        

        
    latencyTest(servers_rdma[0]);



    


    

    for (int sock: servers) {
        if (sock >= 0)
            destroyConnection(sock);
    }
    destroyConnection(master);
    return 0;
}

